from confluent_kafka import Consumer, KafkaError
import time

# 配置Kafka消费者
kafka_config = {
    'bootstrap.servers': '1.94.147.80:9092',  # 替换为你的Kafka服务器地址
    'group.id': 'my-consumer-group',  # 消费者组ID
    'auto.offset.reset': 'earliest'  # 从最早的偏移量开始读取
}

# Kafka主题名称
topic = 'test'  # 替换为你实际的Kafka主题名称

# 接收间隔时间（秒）
recv_interval = 0.01


def print_assignment(consumer, partitions):
    print(f'Assigned partitions: {partitions}')


def Consume_Messages():
    # 创建Kafka消费者实例
    consumer = Consumer(kafka_config)

    # 订阅Kafka主题
    consumer.subscribe([topic], on_assign=print_assignment)

    try:
        while True:
            time.sleep(recv_interval)
            msg = consumer.poll(1.0)  # 轮询消息，超时时间为1秒

            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    print(f"End of partition reached {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")
                elif msg.error():
                    print(f"Error: {msg.error()}")
            else:
                # 解码消息并打印
                message = msg.value().decode('utf-8')
                print(f"Received message: {message}")

            # 可以在这里添加额外的处理逻辑
            # 例如，将数据存储到数据库或进行其他操作

    except KeyboardInterrupt:
        pass
    finally:
        # 关闭消费者
        consumer.close()


if __name__ == "__main__":
    Consume_Messages()
